package org.apache.rocketmq.proxy.grpc.v2.producer;

import apache.rocketmq.v2.*;
import com.google.common.collect.Maps;
import com.google.common.hash.Hashing;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import com.google.protobuf.util.Durations;
import com.google.protobuf.util.Timestamps;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcValidator;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.processor.QueueSelector;
import org.apache.rocketmq.proxy.service.route.AddressableMessageQueue;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CompletableFuture;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/17
 * @方法描述：处理生产者发送的消息的活动处理器
 */
public class SendMessageActivity extends AbstractMessingActivity {
    //构造方法
    public SendMessageActivity(MessagingProcessor messagingProcessor, GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：发送消息的方法，因为proxy只是一个代理模块，该模块会接收生产者客户端发送过来的消息，但这些消息并不会存储在proxy模块，而是被proxy模块转发给broker模块 让broker模块处理这些消息
     */
    public CompletableFuture<SendMessageResponse> sendMessage(ProxyContext ctx, SendMessageRequest request) {
        CompletableFuture<SendMessageResponse> future = new CompletableFuture<>();
        try {
            //判断生产者发送过来的消息是否为空
            if (request.getMessagesCount() <= 0) {
                //如果为空则抛出异常
                throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, "no message to send");
            }
            //得到生产者客户端发送过的消息集合
            List<Message> messageList = request.getMessagesList();
            //得到集合中的第一条消息
            apache.rocketmq.v2.Message message = messageList.get(0);
            //得到消息主题
            Resource topic = message.getTopic();
            //校验topic
            validateTopic(topic);
            future = this.messagingProcessor.sendMessage(
                    ctx,
                    new SendMessageQueueSelector(request),
                    topic.getName(),
                    buildSysFlag(message),
                    buildMessage(ctx, request.getMessagesList(), topic)
            ).thenApply(result -> convertToSendMessageResponse(ctx, request, result));
        } catch (Exception e) {
            future.completeExceptionally(e);
        }
        return future;
    }

    private List<org.apache.rocketmq.common.message.Message> buildMessage(ProxyContext context, List<Message> protoMessageList, Resource topic) {
        String topicName = topic.getName();
        List<org.apache.rocketmq.common.message.Message> messageExtList = new ArrayList<>();
        for (apache.rocketmq.v2.Message protoMessage : protoMessageList) {
            //验证要转换的每一条消息的主题是否和传入的参数一致
            if (!protoMessage.getTopic().equals(topic)) {
                throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, "topic in message is not same");
            }
            //构建单条消息并添加到列表中
            //这里主题信息就是生产者组的名称
            messageExtList.add(buildMessage(context, protoMessage, topicName));
        }
        return messageExtList;
    }

    private org.apache.rocketmq.common.message.Message buildMessage(ProxyContext context, Message protoMessage, String producerGroup) {
        //得到主题名称
        String topicName = protoMessage.getTopic().getName();
        //验证消息大小是否合规
        validateMessageBodySize(protoMessage.getBody());
        //创建消息对象
        org.apache.rocketmq.common.message.Message messageExt = new org.apache.rocketmq.common.message.Message();
        //设置消息主题
        messageExt.setTopic(topicName);
        //设置消息内容
        messageExt.setBody(protoMessage.getBody().toByteArray());
        //构建消息额外属性
        Map<String, String> messageProperty = this.buildMessageProperty(context, protoMessage, producerGroup);
        //设置消息额外属性
        MessageAccessor.setProperties(messageExt, messageProperty);
        return messageExt;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建消息属性的方法
     */
    protected Map<String, String> buildMessageProperty(ProxyContext context, apache.rocketmq.v2.Message message, String producerGroup) {
        //用户自定义的属性的字节大小
        long userPropertySize = 0;
        //得到proxy的配置信息对象
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //创建rocketmq内部消息对象，用于存储用户自定义属性
        org.apache.rocketmq.common.message.Message messageWithHeader = new org.apache.rocketmq.common.message.Message();
        //得到用户自定义的属性
        Map<String, String> userProperties = message.getUserPropertiesMap();
        //检查用户自定义属性数量是否超过限制
        if (userProperties.size() > config.getUserPropertyMaxNum()) {
            throw new GrpcProxyException(Code.MESSAGE_PROPERTIES_TOO_LARGE, "too many user properties, max is " + config.getUserPropertyMaxNum());
        }
        //遍历处理每个用户自定义属性
        for (Map.Entry<String, String> userPropertiesEntry : userProperties.entrySet()) {
            //判断是否使用了系统内部的属性键，也就是说用户自定义的属性的key不能和系统内置的属性key相同
            if (MessageConst.STRING_HASH_SET.contains(userPropertiesEntry.getKey())) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, "property is used by system: " + userPropertiesEntry.getKey());
            }
            //判断属性key是否包含java控制字符
            if (GrpcValidator.getInstance().containControlCharacter(userPropertiesEntry.getKey())) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, "the key of property cannot contain control character");
            }
            //判断属性key是否包含java控制字符
            if (GrpcValidator.getInstance().containControlCharacter(userPropertiesEntry.getValue())) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, "the value of property cannot contain control character");
            }
            //计算属性key的字节大小
            userPropertySize += userPropertiesEntry.getKey().getBytes(StandardCharsets.UTF_8).length;
            //计算属性value的字节大小
            userPropertySize += userPropertiesEntry.getValue().getBytes(StandardCharsets.UTF_8).length;
        }
        //将用户自定义的属性设置到消息对象的属性中
        MessageAccessor.setProperties(messageWithHeader, Maps.newHashMap(userProperties));
        //得到消息的标签
        String tag = message.getSystemProperties().getTag();
        //校验标签的合法性
        GrpcValidator.getInstance().validateTag(tag);
        //设置消息标签到消息对象中
        messageWithHeader.setTags(tag);
        //把消息标签的字节大小累加到用户属性字节总大小中
        userPropertySize += tag.getBytes(StandardCharsets.UTF_8).length;
        //获取消息的key列表
        List<String> keysList = message.getSystemProperties().getKeysList();
        //遍历处理每个key
        for (String key : keysList) {
            //校验key的合法性
            validateMessageKey(key);
            //把key的字节大小累加到用户属性字节总大小中
            userPropertySize += key.getBytes(StandardCharsets.UTF_8).length;
        }
        //把消息的所有key都设置到到消息对象中
        if (keysList.size() > 0) {
            messageWithHeader.setKeys(keysList);
        }
        //在这里检查用户属性总大小是否超过了限制，超过则抛出异常
        if (userPropertySize > config.getMaxUserPropertySize()) {
            throw new GrpcProxyException(Code.MESSAGE_PROPERTIES_TOO_LARGE, "the total size of user property is too large, max is " + config.getMaxUserPropertySize());
        }
        //得到消息Id
        String messageId = message.getSystemProperties().getMessageId();
        //校验消息Id不能为空
        if (StringUtils.isBlank(messageId)) {
            throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_ID, "message id cannot be empty");
        }
        //把消息唯一Id设置为消息属性，属性的key就表示消息的唯一标识符
        MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, messageId);
        //得到消息类型
        MessageType messageType = message.getSystemProperties().getMessageType();
        //判断当前消息是否为事务消息
        if (messageType.equals(MessageType.TRANSACTION)) {
            //如果是事务消息则设置事务消息标志
            MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
            //判断是否设置了孤儿事务恢复时间
            if (message.getSystemProperties().hasOrphanedTransactionRecoveryDuration()) {
                //将Duration类型的恢复时间转换为秒
                long transactionRecoverySecond = Durations.toSeconds(message.getSystemProperties().getOrphanedTransactionRecoveryDuration());
                //判断事务恢复时间是否在合法范围内
                validateTransactionRecoverySecond(transactionRecoverySecond);
                //设置事务消息的检查免疫时间到消息对象的属性中
                MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, String.valueOf(transactionRecoverySecond));
            }
        }
        //设置延迟消息相关属性到消息对象的属性中
        fillDelayMessageProperty(message, messageWithHeader);
        //得到消息失败重试的次数
        int reconsumeTimes = message.getSystemProperties().getDeliveryAttempt();
        //设置消息重试次数到消息对象的属性中
        MessageAccessor.setReconsumeTime(messageWithHeader, String.valueOf(reconsumeTimes));
        //设置生产者分组属性到消息对象的属性中
        MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_PRODUCER_GROUP, producerGroup);
        //得到消息分组信息，如果消息有分组，则意味着是顺序消息
        String messageGroup = message.getSystemProperties().getMessageGroup();
        if (StringUtils.isNotEmpty(messageGroup)) {
            //校验消息分组的合法性
            validateMessageGroup(messageGroup);
            //将消息分组设置为分片key的属性，这样可以保证同一分组的消息被投递到同一队列，这样就可以让消费者顺序消费消息了
            MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_SHARDING_KEY, messageGroup);
        }
        //得到消息追踪上下文
        String traceContext = message.getSystemProperties().getTraceContext();
        //如果追踪上下文不为空，设置到消息对象的属性中
        if (!traceContext.isEmpty()) {
            MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_TRACE_CONTEXT, traceContext);
        }
        //得到消息产生的源主机地址
        String bornHost = message.getSystemProperties().getBornHost();
        //如果源主机地址为空，则使用客户端的远程地址作为源主机
        if (StringUtils.isBlank(bornHost)) {
            bornHost = context.getRemoteAddress();
        }
        //如果源主机地址不为空，将其设置到消息对象的属性中
        if (StringUtils.isNotBlank(bornHost)) {
            MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_BORN_HOST, bornHost);
        }
        //得到消息产生的时间戳
        Timestamp bornTimestamp = message.getSystemProperties().getBornTimestamp();
        //校验时间戳是否有效
        if (Timestamps.isValid(bornTimestamp)) {
            //有效则把消息生产时间设置到消息对象的属性中
            MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_BORN_TIMESTAMP, String.valueOf(Timestamps.toMillis(bornTimestamp)));
        }
        //返回消息属性对象
        return messageWithHeader.getProperties();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：校验消息分组是否合法的方法
     */
    protected void validateMessageGroup(String messageGroup) {
        if (StringUtils.isNotEmpty(messageGroup)) {
            if (StringUtils.isBlank(messageGroup)) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_GROUP, "message group cannot be the char sequence of whitespace");
            }
            int maxSize = ConfigurationManager.getProxyConfig().getMaxMessageGroupSize();
            if (maxSize <= 0) {
                return;
            }
            if (messageGroup.getBytes(StandardCharsets.UTF_8).length >= maxSize) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_GROUP, "message group exceed the max size " + maxSize);
            }
            if (GrpcValidator.getInstance().containControlCharacter(messageGroup)) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_GROUP, "message group cannot contain control character");
            }
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：设置延迟消息相关属性到消息对象的属性中的方法
     */
    protected void fillDelayMessageProperty(apache.rocketmq.v2.Message message, org.apache.rocketmq.common.message.Message messageWithHeader) {
        //判断消息是否设置了投递时间戳
        if (message.getSystemProperties().hasDeliveryTimestamp()) {
            //得到消息投递时间戳
            Timestamp deliveryTimestamp = message.getSystemProperties().getDeliveryTimestamp();
            //将Protobuf的Timestamp转换为毫秒时间戳
            long deliveryTimestampMs = Timestamps.toMillis(deliveryTimestamp);
            //校验延迟时间是否在允许范围内
            validateDelayTime(deliveryTimestampMs);
            //得到proxy的配置信息对象
            ProxyConfig config = ConfigurationManager.getProxyConfig();
            //检查是否启用了延迟级别功能
            if (config.isUseDelayLevel()) {
                //如果启用了则根据投递时间计算对应的延迟级别
                int delayLevel = config.computeDelayLevel(deliveryTimestampMs);
                //设置延迟级别到消息对象的属性中
                MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_DELAY_TIME_LEVEL, String.valueOf(delayLevel));
            }
            //将投递时间戳转换为字符串
            String timestampString = String.valueOf(deliveryTimestampMs);
            //设置定时投递时间戳属性到消息对象属性中
            MessageAccessor.putProperty(messageWithHeader, MessageConst.PROPERTY_TIMER_DELIVER_MS, timestampString);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：校验消息延迟时间是否合法的方法
     */
    protected void validateDelayTime(long deliveryTimestampMs) {
        //得到配置信息中的最大延迟时间
        long maxDelay = ConfigurationManager.getProxyConfig().getMaxDelayTimeMills();
        //如果最大延迟时间配置小于等于0，表示不限制延迟时间
        if (maxDelay <= 0) {
            return;
        }
        //计算消息延迟时间是否超过最大限制
        if (deliveryTimestampMs - System.currentTimeMillis() > maxDelay) {
            throw new GrpcProxyException(Code.ILLEGAL_DELIVERY_TIME, "the max delay time of message is too large, max is " + maxDelay);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：验证消息体大小的方法
     */
    protected void validateMessageBodySize(ByteString body) {
        //得到配置信息中的最大消息大小
        int max = ConfigurationManager.getProxyConfig().getMaxMessageSize();
        //如果最大值小于等于0，表示不限制大小
        if (max <= 0) {
            return;
        }
        //判断消息体大小是否超过限制
        if (body.size() > max) {
            //超过则抛出异常
            throw new GrpcProxyException(Code.MESSAGE_BODY_TOO_LARGE, "message body cannot exceed the max " + max);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：校验事务消息恢复时间是否合法的方法
     */
    protected void validateTransactionRecoverySecond(long transactionRecoverySecond) {
        //得到系统配置的最大事务恢复时间
        long maxTransactionRecoverySecond = ConfigurationManager.getProxyConfig().getMaxTransactionRecoverySecond();
        //如果最大事务恢复时间配置小于等于0，表示不限制恢复时间
        if (maxTransactionRecoverySecond <= 0) {
            return;
        }
        //判断事务恢复时间是否超过最大限制
        if (transactionRecoverySecond > maxTransactionRecoverySecond) {
            throw new GrpcProxyException(Code.BAD_REQUEST, "the max transaction recovery time of message is too large, max is " + maxTransactionRecoverySecond);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：校验消息key合法性的方法
     */
    protected void validateMessageKey(String key) {
        if (StringUtils.isNotEmpty(key)) {
            if (StringUtils.isBlank(key)) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_KEY, "key cannot be the char sequence of whitespace");
            }
            if (GrpcValidator.getInstance().containControlCharacter(key)) {
                throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_KEY, "key cannot contain control character");
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：在第十二版本代码中，该方法暂时不添加注释
     */
    protected SendMessageResponse convertToSendMessageResponse(ProxyContext ctx, SendMessageRequest request,
                                                               List<SendResult> resultList) {
        SendMessageResponse.Builder builder = SendMessageResponse.newBuilder();

        Set<Code> responseCodes = new HashSet<>();
        for (SendResult result : resultList) {
            SendResultEntry resultEntry;
            switch (result.getSendStatus()) {
                case FLUSH_DISK_TIMEOUT:
                    resultEntry = SendResultEntry.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.MASTER_PERSISTENCE_TIMEOUT, "send message failed, sendStatus=" + result.getSendStatus()))
                            .build();
                    break;
                case FLUSH_SLAVE_TIMEOUT:
                    resultEntry = SendResultEntry.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.SLAVE_PERSISTENCE_TIMEOUT, "send message failed, sendStatus=" + result.getSendStatus()))
                            .build();
                    break;
                case SLAVE_NOT_AVAILABLE:
                    resultEntry = SendResultEntry.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.HA_NOT_AVAILABLE, "send message failed, sendStatus=" + result.getSendStatus()))
                            .build();
                    break;
                case SEND_OK:
                    resultEntry = SendResultEntry.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
                            .setOffset(result.getQueueOffset())
                            .setMessageId(StringUtils.defaultString(result.getMsgId()))
                            .setTransactionId(StringUtils.defaultString(result.getTransactionId()))
                            .build();
                    break;
                default:
                    resultEntry = SendResultEntry.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.INTERNAL_SERVER_ERROR, "send message failed, sendStatus=" + result.getSendStatus()))
                            .build();
                    break;
            }
            builder.addEntries(resultEntry);
            responseCodes.add(resultEntry.getStatus().getCode());
        }
        if (responseCodes.size() > 1) {
            builder.setStatus(ResponseBuilder.getInstance().buildStatus(Code.MULTIPLE_RESULTS, Code.MULTIPLE_RESULTS.name()));
        } else if (responseCodes.size() == 1) {
            Code code = responseCodes.stream().findAny().get();
            builder.setStatus(ResponseBuilder.getInstance().buildStatus(code, code.name()));
        } else {
            builder.setStatus(ResponseBuilder.getInstance().buildStatus(Code.INTERNAL_SERVER_ERROR, "send status is empty"));
        }
        return builder.build();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建消息系统标志的方法
     */
    protected int buildSysFlag(apache.rocketmq.v2.Message protoMessage) {
        int sysFlag = 0;
        //获取消息体编码方式
        Encoding bodyEncoding = protoMessage.getSystemProperties().getBodyEncoding();
        //如果使用GZIP压缩，设置压缩标志位
        if (bodyEncoding.equals(Encoding.GZIP)) {
            sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
        }
        //获取消息类型
        MessageType messageType = protoMessage.getSystemProperties().getMessageType();
        //如果是事务消息，设置事务准备标志位
        if (messageType.equals(MessageType.TRANSACTION)) {
            sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
        }
        return sysFlag;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息发送的队列选择器
     */
    protected static class SendMessageQueueSelector implements QueueSelector {

        //发送消息的请求对象
        private final SendMessageRequest request;

        public SendMessageQueueSelector(SendMessageRequest request) {
            this.request = request;
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：选择具体队列发送消息的方法，真正的负载均衡就是在该方法中执行的
         */
        @Override
        public AddressableMessageQueue select(ProxyContext ctx, MessageQueueView messageQueueView) {
            try {
                //得到请求中的第一条消息
                apache.rocketmq.v2.Message message = request.getMessages(0);
                String shardingKey = null;
                //如果请求中只有一条消息，获取其消息分组作为分片键
                if (request.getMessagesCount() == 1) {
                    shardingKey = message.getSystemProperties().getMessageGroup();
                }
                AddressableMessageQueue targetMessageQueue;
                //判断消息是否存在分片键，这里的分片key其实就是生产者组名称
                if (StringUtils.isNotEmpty(shardingKey)) {
                    //获取该消息所属主题下所有可写队列
                    List<AddressableMessageQueue> writeQueues = messageQueueView.getWriteSelector().getQueues();
                    //使用一致性哈希算法计算队列索引
                    int bucket = Hashing.consistentHash(shardingKey.hashCode(), writeQueues.size());
                    //获取目标队列，如果是顺序消息，那么所有消息的生产者组都是一致的
                    //这样一来所有消息就会被投递到同一个队列中，被消费者顺序消费
                    targetMessageQueue = writeQueues.get(bucket);
                } else {
                    //没有分片key时，意味着不是顺序消息，那就使用轮询方式选择目标队列集合
                    targetMessageQueue = messageQueueView.getWriteSelector().selectOneByPipeline(false);
                }
                //返回选择的目标队列
                return targetMessageQueue;
            } catch (Exception e) {
                return null;
            }
        }
    }

}

